package com.lagou.flume;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.compress.utils.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.junit.Test;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class LogTypeInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        //获取event body
        String eventBody = new String(event.getBody(), Charsets.UTF_8);
        //获取 event header
        Map<String, String> headersMap = event.getHeaders();
        //解析body 获取json
        String[] bodyArr = eventBody.split("\\s+");
        try {
            String jsonStr = bodyArr[6];
            // 解析json串获取时间戳
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            String timestampStr = "";
            if (headersMap.getOrDefault("logtype", "").equals("start")) {
                // 取启动日志的时间戳
                timestampStr = jsonObject.getJSONObject("app_active").getString("time");
            } else if (headersMap.getOrDefault("logtype", "").equals("event")) {
                // 取事件日志第一条记录的时间戳
                JSONArray jsonArray = jsonObject.getJSONArray("lagou_event");
                if (jsonArray.size() > 0) {
                    timestampStr = jsonArray.getJSONObject(0).getString("time");
                }
            }

            // 将时间戳转换为字符串 "yyyy-MM-dd"
            // 将字符串转换为Long
            long timestamp = Long.parseLong(timestampStr);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            Instant instant = Instant.ofEpochMilli(timestamp);
            LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
            String date = formatter.format(localDateTime);

            // 将转换后的字符串放置header中
             headersMap.put("logtime", date);
             event.setHeaders(headersMap);
        } catch (Exception e) {
            headersMap.put("logtime", "Unknown");
            event.setHeaders(headersMap);
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        List<Event> lstEvent = new ArrayList<>();
        for (Event event : list) {
            Event intercept = intercept(event);
            if (intercept != null) {
                lstEvent.add(intercept);
            }
        }
        return lstEvent;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder{
        private static LogTypeInterceptor logTypeInterceptor;
        static {
            logTypeInterceptor=new LogTypeInterceptor();
        }
        @Override
        public Interceptor build() {
            //使用单例优化
            return logTypeInterceptor;
        }

        @Override
        public void configure(Context context) {

        }
    }

    @Test
    public void startJunit() {
        String str = "2020-08-20 12:00:58.411 [main] INFO  com.lagou.ecommerce.AppEvent - {'lagou_event':[{'name':'praise','json':{'id':1,'type':4,'add_time':'1597812917144','userid':5,'target':9},'time':1595338371179}],'attr':{'area':'胶南','uid':'2F10092A10','app_v':'1.1.18','event_type':'common','device_id':'1FB872-9A10010','os_type':'2.9','channel':'WX','language':'chinese','brand':'iphone-7'}}";
        Map<String, String> map = new HashMap<>();
        // new Event
        Event event = new SimpleEvent();
        event.setHeaders(map);
        event.setBody(str.getBytes(Charsets.UTF_8));
        // 调用interceptor处理event
        LogTypeInterceptor.Builder builder = new LogTypeInterceptor.Builder();
        Interceptor interceptor = builder.build();
        Event outEvent = interceptor.intercept(event);

        // 处理结果
        Map<String, String> headersMap = outEvent.getHeaders();
        System.out.println(JSON.toJSONString(headersMap));
    }
}
